/**
 * f-club.cn
 * Copyright (c) 2009-2012 All Rights Reserved.
 */
package com.mtoolkit.socket.server;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

import com.mtoolkit.socket.SocketRequest;
import com.mtoolkit.socket.SocketResponse;
import com.mtoolkit.socket.transocder.DefaultSocketRequestTranscoder;
import com.mtoolkit.socket.transocder.DefaultSocketResponseTranscoder;
import com.mtoolkit.socket.transocder.SocketRequestTranscoder;
import com.mtoolkit.socket.transocder.SocketResponseTranscoder;
import com.mtoolkit.util.LogUtil;

public class NioSocketServer {
    
    private InetSocketAddress _address;
    private static Selector _eventSelector;
	
    private NioServerEventNotifier _serverEventNotifier;
	private SocketRequestTranscoder _requestTranscoder = new DefaultSocketRequestTranscoder();
	private SocketResponseTranscoder _responseTranscoder = new DefaultSocketResponseTranscoder();
    
    private int _threadPoolSize;
    private static final int EOF = -1;
    private static final int DEFAULT_THREAD_POOL_SIZE = 4;

    // 响应池
    private static final List<SelectionKey> WPOOL = new LinkedList<SelectionKey>();
    private static final LogUtil LOGGER = LogUtil.getLogger(NioSocketServer.class);
    
    // ---- constructors -----------------------------------------------------------------------
	public NioSocketServer(int port) {
		this(port, DEFAULT_THREAD_POOL_SIZE);
	}
	
	public NioSocketServer(int port, int threadPoolSize) {
		this(new InetSocketAddress(port), threadPoolSize);
	}
	
	public NioSocketServer(InetSocketAddress address, int threadPoolSize) {
		_address = address;
		_threadPoolSize = threadPoolSize;
		
		// 构造事件触发器
        _serverEventNotifier = NioServerEventNotifier.getNotifier();

        // 创建无阻塞网络套接
        try {
			_eventSelector = Selector.open();
			ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
	        serverSocketChannel.configureBlocking(false);
	        ServerSocket serverSocket = serverSocketChannel.socket();
	        serverSocket.bind(address);
	        serverSocketChannel.register(_eventSelector, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
        	LOGGER.error("NioSocketServer occurs I/O exception while building.", e);
        }
	}
	
	// ---- public methods ---------------------------------------------------------------------
	public NioServerEventNotifier getServerEventNotifier() {
		return _serverEventNotifier;
	}
	
	public SocketRequestTranscoder getSocketRequestTranscoder() {
		return _requestTranscoder;
	}
	
	public void setSocketRequestTranscoder(SocketRequestTranscoder requestTranscoder) {
		_requestTranscoder = requestTranscoder;
	}
	
	public SocketResponseTranscoder getSocketResponseTranscoder() {
		return _responseTranscoder;
	}
	
	public void setSocketResponseTranscoder(SocketResponseTranscoder responseTranscoder) {
		_responseTranscoder = responseTranscoder;
	}
	
	public void startup() {
		// 创建读写线程池
		/*
        for (int i = 0; i < _threadPoolSize; i++) {
            new NioServerReader().start();
            new NioServerWriter().start();
        }*/
        
        // 服务端主线程监听
        new Thread(new MainWorkerThread(), "NioServerSocket-MainThread").start();
	}
	
	public void shutdown() {
		
	}
	
    /**
     * 提交新的客户端写请求于主服务线程的回应池中。
     * @param key 选择键。
     */
    public static void processWriteRequest(SelectionKey key) {
        synchronized (WPOOL) {
            WPOOL.add(WPOOL.size(), key);
            WPOOL.notifyAll();
        }
        // 解除selector的阻塞状态，以便注册新的通道
        _eventSelector.wakeup();
    }
    
    private class MainWorkerThread implements Runnable {

        @Override
		public void run() {
            LOGGER.info("==> Nio server socket start to listen on {0}", _address.toString());
            
            SelectionKey key = null;
            Iterator<SelectionKey> iterator = null;
            ServerSocketChannel serverChannel = null;
            
            while (true) {
                try {
                    int num = _eventSelector.select();
                    if (num > 0) {
                        iterator = _eventSelector.selectedKeys().iterator();
                        while (iterator.hasNext()) {
                            key = iterator.next();
                            iterator.remove();
                            
                            // 处理IO事件
                            if (key.isValid() && key.isAcceptable()) {
                                // Accept the new connection
                                serverChannel = (ServerSocketChannel) key.channel();
                                _serverEventNotifier.fireOnAccept();
                                
                                // 触发接受连接事件
                                SocketChannel channel = serverChannel.accept();
                                channel.configureBlocking(false);
                                InnerSocketRequest innerRequest = new InnerSocketRequest(channel);
                                _serverEventNotifier.fireOnAccepted(innerRequest);

                                // 注册读操作，以进行下一步的读操作
                                channel.register(_eventSelector, SelectionKey.OP_READ, innerRequest);
                            } else if (key.isValid() && key.isReadable()) {
                                // 提交给读服务线程读取客户端数据
                                SocketChannel channel = (SocketChannel) key.channel();                       
                                SocketRequest request = doReadSocketRequestBytes(channel);
                                
                                InnerSocketRequest innerRequest = (InnerSocketRequest) key.attachment();
                                innerRequest.setRequestId(request.getRequestId());
                                innerRequest.setRequestData(request.getRequestData());
                                _serverEventNotifier.fireOnReaded(innerRequest);
                                
                                channel.register(_eventSelector, SelectionKey.OP_WRITE, innerRequest);
                            } else if (key.isValid() && key.isWritable()) {
                                // 触发onWrite事件
                                InnerSocketRequest innerRequest = (InnerSocketRequest) key.attachment();
                                SocketResponse innerResponse = new InnerSocketResponse(innerRequest.getRequestId());
                                _serverEventNotifier.fireOnWrite(innerRequest, innerResponse);
                                
                                byte[] responseBytes = _responseTranscoder.sealResponse(innerResponse);
                                SocketChannel channel = (SocketChannel) key.channel();
                                doWriteSocketResponseBytes(channel, responseBytes);
                                
                                // 注册读操作，以进行下一步的读操作
                                innerRequest = new InnerSocketRequest(channel);
                                channel.register(_eventSelector, SelectionKey.OP_READ, innerRequest);
                            }
                        }
                    }
                } catch (Exception ex) {
                    _serverEventNotifier.fireOnError(ex);
                    continue;
                }
            }
		}
        
        private SocketRequest doReadSocketRequestBytes(SocketChannel channel) throws IOException, ClassNotFoundException {
            final int capcity = 1024;
            ByteBuffer buffer = ByteBuffer.allocate(capcity);

            int off = 0;
            byte[] datas = new byte[capcity * 10];
            SocketRequest request = null;
            while (true) {
                buffer.clear();
                int read = channel.read(buffer);
                if (read == EOF) {
                	// TODO: client socket output stream closed
                	channel.socket().close();
                	channel.close();
                	_serverEventNotifier.fireOnClosed(null);
                    return null;
                } else {
                	if ((off + read) > datas.length) {
                		datas = grow(datas, capcity * 10);
                	}

	                byte[] buf = buffer.array();
	                System.arraycopy(buf, 0, datas, off, read);
	                off = off + read;
	                
	                byte[] requestBytes = new byte[off];
	                System.arraycopy(datas, 0, requestBytes, 0, off);
	                
	                request = _requestTranscoder.unsealRequest(requestBytes);
	                if (request != null) {
	                	break;
	                }
                }
            }

            return request;
        }
        
        private byte[] grow(byte[] datas, int size) {
            byte[] temp = new byte[datas.length + size];
            System.arraycopy(datas, 0, temp, 0, datas.length);
            return temp;
        }
        
        private void doWriteSocketResponseBytes(SocketChannel channel, byte[] responseBytes) throws IOException {
            ByteBuffer buffer = ByteBuffer.allocate(responseBytes.length);
            buffer.put(responseBytes, 0, responseBytes.length);
            buffer.flip();
            channel.write(buffer);
        }
	
    }
    
    private static class InnerSocketRequest extends SocketRequest {
        // serialize inner class must be static
        private static final long serialVersionUID = -8486788053967177933L;
        
        private String _requestId;
    	private Object _requestData;
    	private SocketChannel _socketChannel;
    	public InnerSocketRequest(SocketChannel socketChannel) {
    	    super(InnerSocketRequest.class.getName());
    		_socketChannel = socketChannel;
		}
    	
    	@Override
    	public String getRequestId() {
    	    return _requestId;
    	}
    	
    	public void setRequestId(String requestId) {
    	    _requestId = requestId;
    	}
    	
		@Override
		public Object getRequestData() {
			return _requestData;
		}
		
		public void setRequestData(Object message) {
			_requestData = message;
		}

		@Override
		public InetAddress getInetAddress() {
			return _socketChannel.socket().getInetAddress();
		}

		@Override
		public int getPort() {
			return _socketChannel.socket().getPort();
		}

		@Override
		public boolean isConnected() {
			return _socketChannel.isConnected();
		}

		@Override
		public boolean isConnectionPending() {
			return _socketChannel.isConnectionPending();
		}

		@Override
		public boolean getKeepAlive() throws SocketException {
			return _socketChannel.socket().getKeepAlive();
		}

		@Override
		public int getSoTimeout() throws SocketException {
			return _socketChannel.socket().getSoTimeout();
		}

		@Override
		public boolean getTcpNoDelay() throws SocketException {
			return _socketChannel.socket().getTcpNoDelay();
		}

		@Override
		public boolean isClosed() {
			return _socketChannel.socket().isClosed();
		}
    	
    }

    private static class InnerSocketResponse extends SocketResponse {
        // serialize inner class must be static
        private static final long serialVersionUID = -2535994048338108435L;
        
        private String _requestId;
		private Object _responseData;
		
        public InnerSocketResponse(String requestId) {
            _requestId = requestId;
        }
		
		@Override
		public String getRequestId() {
		    return _requestId;
		}
		
		public Object getResponseData() {
		    return _responseData;
		}
		
		public void setResponseData(Object result) {
			_responseData = result;
		}
		
	}
    
    
}
